package com.lmax.disruptor.custom.group2;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class Main
{
    public static void main(final String[] args) throws Exception
    {
        //创建ringBuffer
//        RingBuffer<Order> ringBuffer = RingBuffer.createMultiProducer(Order.INSTANCE, 1024 * 1024, new YieldingWaitStrategy());
        //创建SequenceBarrier
//        SequenceBarrier barriers = ringBuffer.newBarrier();
        //创建3个消费者实例
        Consumer[] consumers = new Consumer[3];
        for (int i = 0; i < consumers.length; i++)
        {
            consumers[i] = new Consumer("c" + i);
        }
        //这一步的目的是把消费者的位置信息引用注入到生产者 如果只有一个消费者的情况可以省略。
        //workerPool.getWorkerSequences()获取Sequence集合
        Disruptor<Order> disruptor = new Disruptor<>(Order.INSTANCE, 2,
                DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWith(consumers);
        RingBuffer<Order> ringBuffer = disruptor.start();
        //创建线程池
        final CountDownLatch latch = new CountDownLatch(1);
        AtomicLong atomicLong = new AtomicLong(0L);
        for (int i = 0; i < 100; i++)
        {
            final Producer producer = new Producer(ringBuffer);
            final int finalI = i;
            new Thread(() ->
            {
                try
                {
                    //等待生产者100个线程启动
                    latch.await();
                    for (int j = 0; j < 100; j++)
                    {
                        //生产数据
//                        System.out.println("i--- " + finalI + ", j---" + j);
                        producer.onData(UUID.randomUUID().toString());
                    }
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }, "thread--" + i).start();
        }
        //等待两秒，等生产者的100个线程启动
        Thread.sleep(2000);
        System.out.println("---------------开始生产-----------------");
        latch.countDown();
        Thread.sleep(5000);
        disruptor.shutdown();
        System.out.println("总数:" + consumers[0].getCount());
    }

    static class IntEventExceptionHandler implements ExceptionHandler<Order>
    {

        @Override
        public void handleEventException(final Throwable arg0, final long arg1, final Order arg2)
        {
            System.out.println(arg0.getMessage());
            System.out.println("arg1---" + arg1 + "--arg2--" + arg2);
        }

        @Override
        public void handleOnShutdownException(final Throwable arg0)
        {
            System.out.println("handleOnShutdownException---" + arg0.getMessage());
        }

        @Override
        public void handleOnStartException(final Throwable arg0)
        {
            System.out.println("handleOnStartException---" + arg0.getMessage());
        }

    }
}